package com.atguigu.wc

import org.apache.flink.streaming.api.scala._

/**
 * @description: flink流处理案例,flink每一个job或步骤都可以设置并行度,并行度不同的时候会根据re-balance来顺序分发
 * @time: 2020/6/17 23:27
 * @author: baojinlong
 **/
object StreamWordCount0 {
  def main(args: Array[String]): Unit = {
    // 创建流处理执行环境,一个任务就需要一个slot.如果并行的时候就需要两个
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 接收socket文件流 linux下: nc  -lk 7777
    val inputDataStream: DataStream[String] = environment.socketTextStream("localhost", 7777)
    // 定义转换操作,设置slot共享组后不同编号的需要在不同的slot里面,并行的走另外一个slot,如flatMap使用2slot,并行度2
    inputDataStream
      .flatMap(_.split(" ")) // 以空格分词,打散得到所有word
      .filter(_.nonEmpty)
      .map((_, 1)) // 转换成(word,count)二元组
      .keyBy(0) // 按照第一个元素分组
      .sum(1) // 按照第二个元素求和

    // 正在等待事件来触发驱动
    environment.execute("stream word count")
  }
}
